package com.bw.sparkstreaming.job2

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * 从hdfs数据源加载数据，这个案例作为了解就可以了
  */
object WordCountForHDFS {
  def main(args: Array[String]): Unit = {
    //设置了日志的级别
    Logger.getLogger("org").setLevel(Level.INFO)
    val conf = new SparkConf().setMaster("local[2]").setAppName(this.getClass.getSimpleName)
    val ssc = new StreamingContext(conf,Seconds(5))

    /**
      * 如果你的地址是一个高可用的一个地址。
      * core-site.xml
      * hdfs-site.xml
      * 复制到maven项目的 resources目录
      *
      * hdfs://localhost:8020//
      */
    val dataDStream = ssc.textFileStream("hdfs://192.168.134.130:8020//sparkstreaming//hdfs")

    val reuslt = dataDStream.flatMap(_.split(","))
      .map((_, 1))
      .reduceByKey(_ + _)

    reuslt.print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }
}
